package com.kingsoft.dc.khaos.util

import java.io._
import java.util.Properties
import java.util.Base64

import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods.parse

import scala.collection.mutable._
import scala.collection.{Map, mutable}
import scala.collection.JavaConverters._

/**
  * Created by chengguangqing on 2019/5/27.
  */
object KhaosUtils {

  /** Returns the system properties map that is thread-safe to iterator over. It gets the
    * properties which have been set explicitly, as well as those for which only a default value
    * has been defined. */
  def getSystemProperties: Map[String, String] = {
    System.getProperties.stringPropertyNames().asScala.map(key => (key, System.getProperty(key))).toMap
  }


  /** Load properties present in the given file. */
  def getPropertiesFromFile(filename: String): Map[String, String] = {
    val file = new File(filename)
    require(file.exists(), s"Properties file $file does not exist")
    require(file.isFile(), s"Properties file $file is not a normal file")

    val inReader = new InputStreamReader(new FileInputStream(file), "UTF-8")
    try {
      val properties = new Properties()
      properties.load(inReader)
      properties.stringPropertyNames().asScala.map(
        k => (k, properties.getProperty(k).trim)).toMap
    } catch {
      case e: IOException =>
        throw new Exception(s"Failed when loading khaos properties from $filename", e)
    } finally {
      inReader.close()
    }
  }


  /** Return the path of the default Khaos properties file. */
  def getDefaultPropertiesFile(env: Map[String, String] = sys.env): String = {
    val khaos = env.get("KHAOS_CONF_DIR")
      .orElse(env.get("KHAOS_HOME").map { t => s"$t${File.separator}conf" })
      .map { t => new File(s"$t${File.separator}khaos-defaults.conf") }
      .filter(_.isFile)
      .map(_.getAbsolutePath)
      .orNull
    khaos
  }

  /** 加载默认配置文件 */
  def loadDefaultConf(filename: String):Map[String, String] = {
    var inputStream: InputStream = null
    val properties = new Properties
    try {
      inputStream = this.getClass.getClassLoader.getResourceAsStream(filename)
      properties.load(inputStream)
    }
    catch {
      case e: IOException => {
        throw new Exception(s"Failed when loading khaos properties from $filename", e)
      }
    } finally {
      inputStream.close()
    }
    val propMap: Map[String, String] = properties.stringPropertyNames().asScala.map(
      k => (k, properties.getProperty(k).trim)).toMap
    propMap
  }



  //解决参数中有空格的情况
  def formatMainArge(args: Array[String]): Array[String]={
    var tmpString = new StringBuilder()
    var resArge = ListBuffer[String]()
    for(arg <- args){
      tmpString.append(arg+" ")
    }
    val confkeyword= List("--userConfig")
    val argsString= tmpString.toString()
    val kvIndex = new ListBuffer[Int]()
    //添加索引值
    for(kw <- confkeyword){
      val tmp = argsString.indexOf(kw)
      if(tmp != -1){
        kvIndex += tmp
      }
    }
    sortList(kvIndex)
    var start = 0
    for(ki <- kvIndex){
      if(!ki.equals(0)){
        resArge += argsString.substring(start,ki)
      }
      start = ki
    }
    resArge += argsString.substring(start,argsString.length)
    resArge.toArray
  }

  //对结果进行排序
  def sortList(kindexs:ListBuffer[Int]): Unit ={
    val length = kindexs.length
    for( i <- 0 to length){
      for(j <- 0 to (length - i - 2)){
        if(kindexs(j) > kindexs(j+1)){//if
        val temp = kindexs(j+1)
          kindexs(j+1) = kindexs(j)
          kindexs(j) = temp
        }
      }
    }
  }

  //解码
  def decodeBase64(args:Array[String]):Array[String] = {
    val str = args.apply(0).substring("--userConfig=".length)
    val res = new String(Base64.getDecoder.decode(str))
    Array("--userConfig="+res)
  }

  //解码
  def decodeBase64Json(userConfig:String):String = {
    val res = new String(java.util.Base64.getDecoder.decode(userConfig))
    res
  }

  //转换作业参数名 khaos_module_hive_sink_jdbc_record_count ==> module.hive.sink.jdbc.record.count
  def transformConfigKey(oldKey:String): String ={
    var _key:String=oldKey
    //判断前缀
    if(oldKey.toLowerCase.startsWith("khaos_")){
      //去除前缀
      _key = _key.substring(6)
      //替换 _
      _key = _key.replaceAll("_",".")

    }
    _key
  }

  /**
   *  过滤出用户设置spark的参数
   */
  def getSparkEngineConfigs(optParamsJson: String, notAllowSetConfigs: Array[String ]): Map[String, String] ={
    implicit val formats = DefaultFormats
    val parser = parse(optParamsJson, true)

    val useSetConfigList = parser.extract[List[userParamBean]]
    val finalConfigMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()

    //处理逻辑：用户参数分为预定义和自定义，如果参数存在重复时，value值相同则合并，不相同则抛错
    for (configEntry <- useSetConfigList){
      //黑名单
      if(notAllowSetConfigs.map(_.toLowerCase).contains(configEntry.params.trim.toLowerCase)){
        throw new Exception(s"Contains illegal parameters![${configEntry.params}]")
      }
      //不存在则添加
      if(!finalConfigMap.keys.toList.contains(configEntry.params.trim)){
        if(configEntry.params.trim.toLowerCase.startsWith("spark.")){
          finalConfigMap.+= ((configEntry.params.trim, configEntry.value.trim))
        }
        if(configEntry.params.trim.toLowerCase.startsWith("hive.")){
          finalConfigMap.+= ((configEntry.params.trim, configEntry.value.trim))
        }
        if(configEntry.params.trim.toLowerCase.startsWith("metastore.")){
          finalConfigMap.+= ((configEntry.params.trim, configEntry.value.trim))
        }
      }else{//存在
        //判断value值是否相同，不相同则抛错，相同则不添加
        if(!finalConfigMap(configEntry.params.trim).toLowerCase.equals(configEntry.value.trim.toLowerCase)){
          throw new Exception(s"Have a same parameters with different values! [${configEntry.params}]")
        }
      }
      //不管是否存在，以特殊前缀开头，都进行设置
      if (configEntry.params.trim.toLowerCase.startsWith("special.param.")){
        finalConfigMap.+=((configEntry.params.trim.substring(14),configEntry.value.trim))
      }
    }
    finalConfigMap
  }

  case class userParamBean(params: String, `type`: String, value: String)

  /**
   * 过滤出插件相关参数
   * @param optParamsJson
   * @param notAllowSetConfigs
   * @return
   */
  def getPluginsConfigs(optParamsJson: String,notAllowSetConfigs: Array[String]): Map[String, String]={
    implicit val formats = DefaultFormats
    val parser = parse(optParamsJson, true)

    val useSetConfigList: List[userParamBean] = parser.extract[List[userParamBean]]
    val finalConfigMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()

    //处理逻辑：用户参数分为预定义和自定义，如果参数存在重复时，value值相同则合并，不相同则抛错
    for (configEntry <- useSetConfigList){
      //黑名单
      if(notAllowSetConfigs.map(_.toLowerCase).contains(configEntry.params.trim.toLowerCase)){
        throw new Exception(s"Contains illegal parameters! [${configEntry.params}]")
      }
      //不存在则添加
      if (!finalConfigMap.keys.toList.contains(configEntry.params.trim)) {
        if(configEntry.params.trim.toLowerCase.startsWith("module.")) {
          finalConfigMap.+=((configEntry.params.trim,configEntry.value.trim))
        }
        if (configEntry.params.trim.toLowerCase.startsWith("common.")){
          finalConfigMap.+=((configEntry.params.trim,configEntry.value.trim))
        }
      } else {  //存在
        //判断value的指是否相同，不相同抛错，相同则不添加
        if(!finalConfigMap(configEntry.params.trim).toLowerCase.equals(configEntry.value.trim.toLowerCase)) {
          throw  new Exception(s"Have a same parameters with different values![${configEntry.params}]")
        }
      }
      if (configEntry.params.trim.toLowerCase.startsWith("special.param.")) {
        finalConfigMap.+=((configEntry.params.trim.substring(14),configEntry.value.trim))
      }
    }
    finalConfigMap
  }
}

